package com.dataseek.iot.mqtt.config;

import com.dataseek.iot.mqtt.callback.PubCallBack;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * 订阅MQTT数据
 */
@Component
public class MqttSub {
    @Value("${mqtt.url}")
    private  String HOST;

    @Value("${mqtt.subscriber.clientId}")
    private  String clientid;

    @Value("${mqtt.subscriber.defaultTopic}")
    private  String topic;

    @Value("${mqtt.username}")
    private String username;

    @Value("${mqtt.password}")
    private String password;

    public MqttClient client;

    /**
     * true为非持久订阅
     * <p>
     * 方法实现说明 断线重连方法，如果是持久订阅，重连是不需要再次订阅，如果是非持久订阅，重连是需要重新订阅主题 取决于options.setCleanSession(true);
     * <p>
     * 就是这里的clientId，服务器用来区分用户的，不能重复,clientId不能和发布的clientId一样，否则会出现频繁断开连接和重连的问题
     * 不仅不能和发布的clientId一样，而且也不能和其他订阅的clientId一样，如果想要接收之前的离线数据，这就需要将client的 setCleanSession
     * 设置为false，这样服务器才能保留它的session，再次建立连接的时候，它就会继续使用这个session了。 这时此连接clientId 是不能更改的。
     * 但是其实还有一个问题，就是使用热部署的时候还是会出现频繁断开连接和重连的问题，可能是因为刚启动时的连接没断开，然后热部署的时候又进行了重连，重启一	  *   下就可以了
     * System.currentTimeMillis()
     *
     * @throws MqttException
     */
    public void connect() throws MqttException {
        //防止重复创建MQTTClient实例
        if (client == null) {
            System.out.println("订阅者第一次链接");
            // MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(HOST, clientid, new MemoryPersistence());
            //如果是订阅者则添加回调类，发布不需要
            client.setCallback(new PubCallBack(MqttSub.this));
        }
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(10);
        //设置心跳
        options.setKeepAliveInterval(20);
        //判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
        if (!client.isConnected()) {
            client.connect(options);
            System.out.println(HOST+"=============MQtt 连接成功!!!============"+clientid);
        } else {
            client.disconnect();
            client.connect(options);
            System.out.println(HOST+"=============已断开,从新连接成功!!!============="+clientid);
        }
    }

    /**
     * 初始化订阅信息
     */
    public void init() {
        try {
            //链接MQTT
            connect();
            //订阅消息topic
            System.out.println("subscribe====="+topic);
            subscribe(topic,1);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅某个主题，qos默认为0
     *
     * @param topic .
     */
    public void subscribe(String topic) {
        subscribe(topic, 0);
    }

    /**
     * 订阅某个主题
     *
     * @param topic .
     * @param qos   .
     */
    public void subscribe(String topic, int qos) {
        try {
         System.out.println("订阅topic======="+topic);
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}